package com.eric.log.mapred;

import java.io.IOException;
import java.net.URL;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;

public class SourceMapRed extends VisitMapRed{
	public static class SourceMapper extends Mapper<Object, Text, Text, Text>{
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			String url = value.toString().split(Config.FS)[3];
			
			if (url.startsWith("http") || url.startsWith("https")) {
				URL refUrl = new URL(url);
				Text newkey = new Text(refUrl.toString());
					
				String refDomain = refUrl.getHost();
				String refUri = refUrl.getPath();
				String bounce = value.toString().split(Config.FS)[5];
				String goal = value.toString().split(Config.FS)[6];
				
				
				context.write(newkey, new Text(refDomain+Config.FS+
												refUri+Config.FS+
												bounce+Config.FS+
												goal)
				);
			}
		}
	}
	
	public static class SourceReducer extends Reducer<Text,Text,Text,Text> {
		public void reduce (Text key,Iterable<Text> value,Context context) throws IOException, InterruptedException {
			int goalCounter = 0;
			int bounceCounter = 0;
			int visitCounter = 1;
			for (Text v : value) {
				if (v.toString().split(Config.FS)[2].equals("true")) bounceCounter++;
				if (v.toString().split(Config.FS)[3].equals("true")) goalCounter++;
				visitCounter++;
			}
			context.write(key, new Text(visitCounter+Config.FS+
										bounceCounter+Config.FS+
										goalCounter));
		}
	}
	
	public static void run (String input,String output) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
	    Job job = new Job(conf,"Source Report");
	    
	    job.setMapperClass(SourceMapper.class);
	    job.setReducerClass(SourceReducer.class);
	    
	    job.setJarByClass(SourceMapRed.class);
	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(Text.class);
	    
	    FileInputFormat.addInputPath(job, new Path(input));
	    FileOutputFormat.setOutputPath(job, new Path(output));
	    job.waitForCompletion(true);
	}
}
